package com.we.flink.utils;

import org.apache.flink.api.java.utils.ParameterTool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Iterator;
import java.util.Set;

public class WeKafkaPropertyReader {
    private static ParameterTool param;
    public static Logger LOG = LoggerFactory.getLogger(WeKafkaPropertyReader.class);

    public static WeKafkaPropertyReader init(String propFilePath) {
        try {
            param = getReader(propFilePath);
        } catch (Exception e) {
            LOG.error("Exception: " + e.getMessage());
            e.printStackTrace();
        }
        return new WeKafkaPropertyReader();
    }

    public String getKfkBootStrapServer() {
        return param.get("kfkBootStrapServer");
    }

    public String getKfkTopic() {
        return param.get("kfkTopic");
    }

    public int getKfkPartitions() {
        return Integer.parseInt(param.get("kfkPartitions"));
    }

    public String getTmpKfkBootStrapServer() {
        return param.get("tmpkfkBootStrapServer");
    }

    public String getTmpKfkTopic() {
        return param.get("tmpkfkTopic");
    }

    public int getTmpKfkPartitions() {
        return Integer.parseInt(param.get("tmpkfkPartitions"));
    }

    public String getTmpKfkGroupId() {
        return param.get("tmpkfkGroupId");
    }

    public String getTmpKfkOffset() {
        return param.get("tmpkfkOffset");
    }

    public String getRocksDBBackendUrl() {
        return param.get("rocksDBBackendURL");
    }

    public String getHiveCatlogName() {
        return param.get("hiveCatlogName");
    }

    public String getHiveDBName() {
        return param.get("hiveDBName");
    }

    public String getHiveConfDir() {
        return param.get("hiveConfDir");
    }

    public static ParameterTool getReader(String filePath) throws IOException {
        InputStream resourceAsStream =
                WeKafkaPropertyReader.class.getClassLoader().getResourceAsStream(filePath);
        return ParameterTool.fromPropertiesFile(resourceAsStream);
    }

    public static void main(String[] args) {
        ParameterTool reader = null;
        try {
            reader = WeKafkaPropertyReader.getReader("sql_cmp.properties");
            System.out.println("size: " + reader.getProperties().size());
            Set<Object> keySet = reader.getProperties().keySet();
            Iterator<Object> iterator = keySet.iterator();
            while(iterator.hasNext()){
                Object key = iterator.next();
                //select sout
//                System.out.println(",a."+key+" as " + key+"a");
//                System.out.println(",b."+key+" as " + key+"b");
                //where sout1
//                System.out.println("or "+key+"a"+ " != " + key+"b");
                //where sout2
//                System.out.println("or nvl(a."+key+",0) != nvl(" + "b."+key+",0)");
//                System.out.println(",nvl("+ key + ",0) as " + key);
                //nvl()
//                System.out.println(",nvl(t2."+ key + ", 0) as " + key);
//                System.out.println(",nvl("+ key + ", -999) as " + key);
                //trans
//                System.out.println(","+key + " as phone_call__"+key);
                // sql
                System.out.println("left join (select * from "+key + " where dt = '${dt}') b on a.user_key = b.user_key ");
                // cast
//                System.out.println(key + " as " + key+ "_new,");
                //cast
//                cast(nvl(td_2794845_device_id_p2p,0) as double) as td_l3m_app_cnt_p2p_device_id,
//                System.out.println("cast(nvl("+key+",0) as double) as "+ key+",");
                //cast2
//                case when als_m12_id_af_allnum is null or als_m12_id_af_allnum ='null'then 0 else cast(als_m12_id_af_allnum as double) end as als_m12_id_af_allnum,
//                System.out.println(",case when "+ key + " is null or " + key + " = 'null' then 0 else cast(" + key +" as double) end as "+ key);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        //        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //        System.out.println(parameterTool.getNumberOfParameters());
        //        System.out.println("-hfqaudithomepagelog \n" +
        //                "-lendauditrecord \n" +
        //                "-offlinelabel \n" +
        //                "-risktag \n" +
        //                "-savemoneycardmember \n" +
        //                "-savemoncardmemspecial \n" +
        //                "-supermemrecord \n" +
        //                "-txxyorderrescallback \n" +
        //                "-useraccount \n" +
        //                "-useraudit \n" +
        //                "-usercertifycmpy \n" +
        //                "-usercertifyedu \n" +
        //                "-userentrepreneur \n" +
        //                "-userlevel \n" +
        //                "-userloan \n");
        //        System.out.println(parameterTool.has("testkey"));
        //        System.out.println(parameterTool.has("testv"));

    }
}
